package com.shujia.traffic.dws

import com.shujia.traffic.common.SparkUtil
import com.shujia.traffic.dws.DwsAppVioZqzfDay.day
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object DwsVioWfjlGdTjDay extends SparkUtil {
  /**
   * SparkSQL任务提交的命令：
   * spark-submit --master local --conf spark.sql.shuffle.partitions=2 --class com.shujia.traffic.dws.DwsVioWfjlGdTjDay traffic-1.0-SNAPSHOT.jar 20250425
   */
  override def run(): Unit = {
    val spark: SparkSession = sparkSession

    import org.apache.spark.sql.functions._
    import spark.implicits._

    // 加载数据
    val surveilDF: DataFrame = spark.table("dwd.dwd_base_vio_surveil_msk_d").where($"pt" === day)

    // 创建对应结果表的分区
    spark.sql(
      s"""
         |alter table dws.dws_app_vio_wsjl_gd_tj_d add if not exists partition (pt = '${day}')
         |""".stripMargin)

    // 统计每个设备每一天拍摄的违法记录数
    surveilDF
      .groupBy($"sbbh", substring($"wfsj", 1, 10) as "wfsj_day")
      .agg(count("*") as "sbwfs_day")
      // 使用箱线图原理找出违法记录数过多的异常设备
      .withColumn("rn", row_number() over Window.partitionBy($"sbbh").orderBy($"sbwfs_day"))
      // 统计每个设备有多少条数据
      .withColumn("total_cnt", count("*") over Window.partitionBy($"sbbh"))
      // 分别计算Q1、Q3
      .withColumn("Q1_pos", ceil($"total_cnt" / 4))
      .withColumn("Q3_pos", floor($"total_cnt" * 3 / 4))
      .withColumn("Q1", sum(when($"rn" === $"Q1_pos", $"sbwfs_day").otherwise(0)) over Window.partitionBy($"sbbh"))
      .withColumn("Q3", sum(when($"rn" === $"Q3_pos", $"sbwfs_day").otherwise(0)) over Window.partitionBy($"sbbh"))
      // 计算上边界
      .withColumn("zdgjz",$"Q3"+($"Q3" - $"Q1") * 1.5)
      .where($"sbwfs_day" > $"zdgjz"
        // 找出当天违法记录数过多的异常设备
//        and regexp_replace($"wfsj_day","-","") === day
      )
      .select(
        $"sbbh"
        ,$"sbwfs_day"
        ,$"zdgjz"
        ,$"wfsj_day"
      ).write
      .format("csv")
      .option("sep", "|")
      .mode(SaveMode.Overwrite)
      // 只会自动创建目录，不会创建分区
      .save(s"/project/traffic/dws/dws_app_vio_wsjl_gd_tj_d/pt=${day}")
  }
}
